module Airbrake
  ##
  # Responsible for sending notices to Airbrake asynchronously. The class
  # supports an unlimited number of worker threads and an unlimited queue size
  # (both values are configurable).
  #
  # @see SyncSender
  class AsyncSender
    ##
    # @param [Airbrake::Config] config
    def initialize(config)
      @config = config
      @unsent = SizedQueue.new(config.queue_size)
      @sender = SyncSender.new(config)
      @closed = false
      @workers = ThreadGroup.new
      @mutex = Mutex.new
      @pid = nil
    end

    ##
    # Asynchronously sends a notice to Airbrake.
    #
    # @param [Airbrake::Notice] notice A notice that was generated by the
    #   library
    # @return [nil]
    def send(notice)
      return will_not_deliver(notice) if @unsent.size >= @unsent.max

      @unsent << notice
      nil
    end

    ##
    # Closes the instance making it a no-op (it shut downs all worker
    # threads). Before closing, waits on all unsent notices to be sent.
    #
    # @return [void]
    # @raise [Airbrake::Error] when invoked more than one time
    def close
      threads = @mutex.synchronize do
        if closed?
          raise Airbrake::Error, 'attempted to close already closed sender'
        end

        unless @unsent.empty?
          msg = "#{LOG_LABEL} waiting to send #{@unsent.size} unsent notice(s)..."
          @config.logger.debug(msg + ' (Ctrl-C to abort)')
        end

        @config.workers.times { @unsent << :stop }
        @closed = true
        @workers.list.dup
      end

      threads.each(&:join)
      @config.logger.debug("#{LOG_LABEL} closed")
    end

    ##
    # Checks whether the sender is closed and thus usable.
    # @return [Boolean]
    def closed?
      @closed
    end

    ##
    # Checks if an active sender has any workers. A sender doesn't have any
    # workers only in two cases: when it was closed or when all workers
    # crashed. An *active* sender doesn't have any workers only when something
    # went wrong.
    #
    # Workers are expected to crash when you +fork+ the process the workers are
    # living in. In this case we detect a +fork+ and try to revive them here.
    #
    # Another possible scenario that crashes workers is when you close the
    # instance on +at_exit+, but some other +at_exit+ hook prevents the process
    # from exiting.
    #
    # @return [Boolean] true if an instance wasn't closed, but has no workers
    # @see https://goo.gl/oydz8h Example of at_exit that prevents exit
    def has_workers?
      return false if @closed

      if @pid != Process.pid && @workers.list.empty?
        @pid = Process.pid
        spawn_workers
      end

      !@closed && @workers.list.any?
    end

    private

    def spawn_workers
      @workers = ThreadGroup.new
      @config.workers.times { @workers.add(spawn_worker) }
      @workers.enclose
    end

    def spawn_worker
      Thread.new do
        while (notice = @unsent.pop) != :stop
          @sender.send(notice)
        end
      end
    end

    def will_not_deliver(notice)
      backtrace = notice[:errors][0][:backtrace].map do |line|
        "#{line[:file]}:#{line[:line]} in `#{line[:function]}'"
      end
      @config.logger.error(
        "#{LOG_LABEL} AsyncSender has reached its capacity of "                   \
        "#{@unsent.max} and the following notice will not be delivered "          \
        "Error: #{notice[:errors][0][:type]} - #{notice[:errors][0][:message]}\n" \
        "Backtrace: \n" + backtrace.join("\n")
      )
      nil
    end
  end
end
